package com.kira.scaffoldmvc.Disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class DisruptorPerformanceTest {

    // 测试配置
    private static final int EVENT_COUNT = 10_000_000; // 事件总数
    private static final int QUEUE_CAPACITY = 1024;    // 队列容量
    private static final int WARMUP_ROUNDS = 3;        // 预热轮次
    
    public static void main(String[] args) throws InterruptedException {
        System.out.println("性能对比测试 - Disruptor vs BlockingQueue");
        System.out.println("事件总数: " + EVENT_COUNT);
        System.out.println("队列容量: " + QUEUE_CAPACITY);
        System.out.println("预热轮次: " + WARMUP_ROUNDS);
        System.out.println("--------------------------------------");
        
        // 预热JVM
        System.out.println("开始预热...");
        for (int i = 0; i < WARMUP_ROUNDS; i++) {
            testBlockingQueue();
            testDisruptor();
        }
        System.out.println("预热完成\n");
        
        // 正式测试
        System.out.println("测试传统阻塞队列...");
        long queueTime = testBlockingQueue();
        System.out.println("阻塞队列耗时: " + queueTime + " ms\n");
        
        System.out.println("测试Disruptor...");
        long disruptorTime = testDisruptor();
        System.out.println("Disruptor耗时: " + disruptorTime + " ms\n");
        
        // 性能对比
        double throughputRatio = (double) disruptorTime / queueTime;
        System.out.println("性能对比结果:");
        System.out.printf("Disruptor 吞吐量是阻塞队列的 %.2f 倍%n", 1 / throughputRatio);
        System.out.printf("阻塞队列耗时是 Disruptor 的 %.2f 倍%n", throughputRatio);
    }

    // 测试传统阻塞队列性能
    private static long testBlockingQueue() throws InterruptedException {
        final BlockingQueue<Long> queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);
        final AtomicLong counter = new AtomicLong(0);
        
        // 消费者线程
        Thread consumer = new Thread(() -> {
            try {
                while (counter.get() < EVENT_COUNT) {
                    Long value = queue.take();
                    counter.incrementAndGet();
                    // 模拟简单处理
                    long result = value * value;
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        consumer.start();
        
        long startTime = System.currentTimeMillis();
        
        // 生产者
        for (long i = 0; i < EVENT_COUNT; i++) {
            queue.put(i);
        }
        
        // 等待消费完成
        while (counter.get() < EVENT_COUNT) {
            Thread.sleep(1);
        }
        
        long endTime = System.currentTimeMillis();
        
        consumer.interrupt();
        consumer.join();
        
        return endTime - startTime;
    }

    // 测试Disruptor性能
    private static long testDisruptor() throws InterruptedException {
        // 事件工厂
        EventFactory<LongEvent> factory = LongEvent::new;
        
        // 创建Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(
                factory,
                QUEUE_CAPACITY,
                DaemonThreadFactory.INSTANCE,
                ProducerType.SINGLE,
                new YieldingWaitStrategy()
        );
        
        // 计数器用于跟踪处理完成
        final AtomicLong eventCounter = new AtomicLong(0);
        
        // 事件处理器
        EventHandler<LongEvent> handler = (event, sequence, endOfBatch) -> {
            // 模拟简单处理
            long result = event.getValue() * event.getValue();
            eventCounter.incrementAndGet();
        };
        
        disruptor.handleEventsWith(handler);
        
        // 启动Disruptor
        RingBuffer<LongEvent> ringBuffer = disruptor.start();
        
        long startTime = System.currentTimeMillis();
        
        // 发布事件
        for (long i = 0; i < EVENT_COUNT; i++) {
            long sequence = ringBuffer.next();
            LongEvent event = ringBuffer.get(sequence);
            event.setValue(i);
            ringBuffer.publish(sequence);
        }
        
        // 等待所有事件处理完成
        while (eventCounter.get() < EVENT_COUNT) {
            Thread.sleep(1);
        }
        
        long endTime = System.currentTimeMillis();
        
        disruptor.shutdown();
        
        return endTime - startTime;
    }

    // 事件类
    static class LongEvent {
        private long value;
        
        public void setValue(long value) {
            this.value = value;
        }
        
        public long getValue() {
            return value;
        }
    }
}